package com.atguigu.realtime.streaming.apps

import java.util
import com.alibaba.fastjson.{JSON, JSONArray, JSONObject}
import com.atguigu.realtime.constants.TopicConstant
import com.atguigu.realtime.streaming.utils.MyKafkaUtil
import com.atguigu.realtime.utils.KafkaClientUtil
import com.google.gson.Gson
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka010.{CanCommitOffsets, HasOffsetRanges, OffsetRange}
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * Created by Smexy on 2022/5/24
 *
 *    熟悉数据:
 *
 *    五种数据: start,actions,displays,page,err
 *
 *    特征：
 *          ①所有的数据都有common(设备的一般信息)
 *          ②都有ts, 如果是actions，取action中的ts，而不是log最外层的ts
 *          ③如果是启动日志，有start，没有page
 *          ④除了启动日志，都有page部分
 *          ⑤actions和displays要炸裂，粒度分解为每个action,display是一行
 *
 *
 *   -------------------------------------
 *      希望采取幂等输出的方式。
 *        目前无法实现幂等输出，无法保证精确一次性。
 *        在后续的消费端，进行去重。
 *
 *        为了效率，只保证at least once
 *-----------------------
 *    解析要求，把所有的数据全部展平!
 *      {
 *        a:{
 *            b:xx,
 *            c:xx
 *        }
 *
 *      }
 *
 *      展平:
 *      {
 *        b:xx,
 *        c:xx
 *      }
 *
 *
 */
object LogDiversionApp extends BaseApp {
  override var appName: String = "LogDiversionApp"
  override var batchDuration: Int = 10
  override var groupId: String = "realtime1227"
  override var topic: String = TopicConstant.ORIGINAL_LOG

  def parseRecord(rdd: RDD[ConsumerRecord[String, String]]):Unit = {

    rdd.foreachPartition(partition => {

      val gson = new Gson()

      partition.foreach(record => {


        //将ConsumerRecord的value部分解析为Map
        val jo: JSONObject = JSON.parseObject(record.value())

        // 把common部分提取出来
        val result: util.Map[String, AnyRef] = JSON.parseObject(jo.getString("common")).getInnerMap

        // 放入ts
        result.put("ts",jo.getLong("ts"))

        //判断当前数据的类型
        if (jo.containsKey("start") && !jo.containsKey("err")){

          //启动日志的解析逻辑
          val startMap: util.Map[String, AnyRef] = JSON.parseObject(jo.getString("start")).getInnerMap

          //两个map合并
          result.putAll(startMap)

          //写入kafka
          KafkaClientUtil.sendDataToKafka(TopicConstant.STARTUP_LOG,gson.toJson(result))

          //剩下四种类型都有page
        }else if(jo.containsKey("page")){

          //page日志的解析逻辑
          val pageMap: util.Map[String, AnyRef] = JSON.parseObject(jo.getString("page")).getInnerMap

          result.putAll(pageMap)

          KafkaClientUtil.sendDataToKafka(TopicConstant.PAGE_LOG,gson.toJson(result))


           if (jo.containsKey("err")){

             //err日志的解析逻辑
             val errMap: util.Map[String, AnyRef] = JSON.parseObject(jo.getString("err")).getInnerMap

             // result已经是合并了页面信息的map集合
             result.putAll(errMap)

             if (jo.containsKey("actions")){
               result.put("actions",jo.getString("actions"))
             }

             if (jo.containsKey("displays")){
               result.put("displays",jo.getString("displays"))
             }

             if (jo.containsKey("start")){
               result.put("start",jo.getString("start"))
             }

             KafkaClientUtil.sendDataToKafka(TopicConstant.ERROR_LOG,gson.toJson(result))


           }else {
             if (jo.containsKey("actions")) {

               //actioins日志的解析逻辑
               parseJsonArray("actions", result, TopicConstant.ACTION_LOG)


             }

             if (jo.containsKey("displays")) {

               //displays日志的解析逻辑
               parseJsonArray("displays", result, TopicConstant.DISPLAY_LOG)

             }
           }

          /**
           *
           * @param `type`: 解析的数组类型，必须是actions 或 displays。
           * @param mergeMap： 要合并的map
           * @param topic: 发送的主题
           */
          def parseJsonArray(`type`:String,mergeMap: util.Map[String, AnyRef],topic:String): Unit ={

            val array: JSONArray = JSON.parseArray(jo.getString(`type`))

            //遍历
            for(i <- 0 until array.size()){

              //取出{}
              val eleMap: util.Map[String, AnyRef] = JSON.parseObject(array.getString(i)).getInnerMap

              mergeMap.putAll(eleMap)

              //写出
              KafkaClientUtil.sendDataToKafka(topic,gson.toJson(mergeMap))

            }



          }

        }

      })

    })
  }



  def main(args: Array[String]): Unit = {

    //构造StreamingContext
     context = new StreamingContext("local[*]", appName, Seconds(batchDuration))

    runApp{

      //传入自己的逻辑
      val ds: InputDStream[ConsumerRecord[String, String]] = MyKafkaUtil.getKafkaStream(Array(topic), context, groupId)

      ds.foreachRDD(rdd => {

        if (!rdd.isEmpty()){

          //获取偏移量
          val ranges: Array[OffsetRange] = rdd.asInstanceOf[HasOffsetRanges].offsetRanges

          //处理
          parseRecord(rdd)

          //提交偏移量
          ds.asInstanceOf[CanCommitOffsets].commitAsync(ranges)

        }

      })
    }
  }
}
